package com.yonyou.findata.kafka.adminop;

import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;

import java.util.Properties;

/**
 * @author pizhihui
 * @date 2020-06-11
 */
public class AdminOpUtils {


    public static void listTopic() {
        ListTopicsResult listTopicsResult = getClient().listTopics();
        System.out.println(JSONObject.toJSONString(listTopicsResult));
    }

    public static void createTopic(String topicName, int numPartitons, int replicationFactor) {
        NewTopic newTopic = new NewTopic(topicName, numPartitons, (short) replicationFactor);
        CreateTopicsResult res = getClient().createTopics(Lists.newArrayList(newTopic));
        System.out.println(JSONObject.toJSONString(res));
    }

    private static AdminClient getClient() {
       return Holder.INSTANCE;
    }

    private static class Holder {
        private static Properties initProperties() {
            Properties props = new Properties();
            props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
            return props;
        }

        private static final AdminClient INSTANCE = AdminClient.create(initProperties());
    }


}
